package com.xueshan.disruptor.lmax.queue;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import com.xueshan.disruptor.lmax.consumer.DisruptorConsumer;
import com.xueshan.disruptor.lmax.event.ObjectEvent;
import com.xueshan.disruptor.lmax.event.ObjectEventFactory;

import java.util.concurrent.Executors;

/**
 * created with IDEA
 *
 * @author:huqm
 * @date:2021/3/18
 * @time:17:58 <p>
 *
 * </p>
 */
public class RingBufferFactory {
    // 创建"点对点模式"的操作队列，即同一事件会被一组消费者其中之一消费
    public static <T> RingBuffer<ObjectEvent> getWorkRingBuffer(int queueSize, boolean isMoreProducer,
                                                               DisruptorConsumer... consumers) {
        RingBuffer<ObjectEvent> ringBuffer = RingBuffer.create(isMoreProducer ? ProducerType.MULTI : ProducerType.SINGLE, new ObjectEventFactory(), queueSize, new YieldingWaitStrategy());
        SequenceBarrier barriers = ringBuffer.newBarrier();
        WorkerPool<ObjectEvent> workerPool = new WorkerPool<ObjectEvent>(ringBuffer, barriers, new EventExceptionHandler(), consumers);
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        workerPool.start(Executors.newFixedThreadPool(6));
        return ringBuffer;
    }
}
